package com.turbo;

import lombok.*;
import lombok.experimental.Accessors;
import org.apache.commons.lang3.StringUtils;

import java.io.*;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author: zwx
 * @time: 2021/4/27 20:37
 */
public class BigFileReader {
    private final byte newLineByte = '\n';
    private final byte enterByte = '\r';

    private final int threadSize;
    private final long fileLength;
    private final int bufferSize;
    private final String charSet;
    private final IHandle ihandle;
    private RandomAccessFile rAccessFile;
    private final ExecutorService executorService;
    private final Set<StartEndPair> startEndPairs;

    /** 性能测试 */
    private CyclicBarrier cyclicBarrier;
    private final AtomicLong countLine = new AtomicLong(0L);

    public BigFileReader(File file, Integer bufferSize, String charSet, Integer threadSize, IHandle handle) {
        this.fileLength = file.length();
        this.bufferSize = bufferSize;
        this.charSet = charSet;
        this.threadSize = threadSize;
        this.ihandle = handle;
        try {
            this.rAccessFile = new RandomAccessFile(file, "r");
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
        this.executorService = new ThreadPoolExecutor(threadSize,
                                                      threadSize,
                                                      0,
                                                      TimeUnit.SECONDS,
                                                      new LinkedBlockingQueue<>(threadSize),
                                                      Executors.defaultThreadFactory(),
                                                      new ThreadPoolExecutor.AbortPolicy());
        this.startEndPairs = new HashSet<>();
    }

    public void start(){
        long perSize = this.fileLength / threadSize;
        try {
            calculateStartEnd(0, perSize);
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 性能测试开始 */
        final long startTime = System.currentTimeMillis();
        cyclicBarrier = new CyclicBarrier(startEndPairs.size(), () -> {
            System.out.println("use time: " + (System.currentTimeMillis() - startTime));
            System.out.println("all line: " + countLine.get());
        });
        // 性能测试结束 */

        for (StartEndPair pair : startEndPairs) {
            executorService.execute(new SliceReaderTask(pair));
        }
    }

    /**
     * 计算分区读取的起始位置与结束位置
     * 保证每个分区的结束位置都是换行或回车的位置，以避免一条数据分成两个部分到不同的线程中
     * @author zwx
     * @param start 起始位置
     * @param perSize 平分处理数据的大小
     * @throws IOException io异常
     */
    private void calculateStartEnd(long start, long perSize) throws IOException {
        if (start > fileLength-1){
            return;
        }
        StartEndPair pair = new StartEndPair();
        pair.start = start;
        long endPosition = start + perSize - 1;
        if (endPosition >= fileLength){
            pair.end = fileLength - 1;
            startEndPairs.add(pair);
            return;
        }
        rAccessFile.seek(endPosition);
        byte tmp = (byte)rAccessFile.read();
        while (tmp != newLineByte && tmp != enterByte){
            endPosition++;
            if (endPosition >= fileLength - 1){
                endPosition = fileLength - 1;
                break;
            }
            rAccessFile.seek(endPosition);
            tmp = (byte)rAccessFile.read();
        }
        pair.end = endPosition;
        startEndPairs.add(pair);

        calculateStartEnd(endPosition + 1, perSize);
    }

    /**
     * 线程处理过程
     * @author zwx
     */
    private class SliceReaderTask implements Runnable {
        private final long start;
        private final long sliceSize;
        private final byte[] readBuff;
        // private Set<Person> personSet;

        public SliceReaderTask(StartEndPair pair){
            this.start = pair.start;
            this.sliceSize = pair.end - pair.start + 1;
            this.readBuff = new byte[bufferSize];
        }

        @Override
        public void run() {
            try {
                MappedByteBuffer mapBuffer = rAccessFile.getChannel().map(FileChannel.MapMode.READ_ONLY, start, sliceSize);
                ByteArrayOutputStream bos = new ByteArrayOutputStream();
                for (int offset = 0; offset < sliceSize; offset += bufferSize) {
                    int readLength;
                    if (offset + bufferSize <= sliceSize){
                        readLength = bufferSize;
                    }else {
                        readLength = (int)sliceSize - offset;
                    }
                    // 将内存映射中的数据读到readBuff中，从readBuff索引0开始存储到索引readLength
                    mapBuffer.get(readBuff, 0, readLength);
                    // 遍历readBuff 将每行的数据写入到bos中，然后传给IHandle处理
                    for (int i = 0; i < readLength; i++) {
                        byte tmp = readBuff[i];
                        if (tmp == newLineByte || tmp == enterByte){
                            handle(bos.toByteArray());
                            bos.reset();
                        }else {
                            bos.write(tmp);
                        }
                    }
                }
                if (bos.size() > 0){
                    handle(bos.toByteArray());
                }
                // 性能测试用，记录完成线程数 */
                cyclicBarrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }

        }

        /**
         * 如果要批量导入到数据库，起始这里的方法可以改成将每一行的数据都封装到对应的集合中
         * 如：person实体，这里可以将每行数据封装到person实体中，然后add到personSet集合中
         * 封装完成之后再调用IHandle，函数式接口去insert数据
         * @author zwx
         * @param bytes 每行数据 字节
         * @throws UnsupportedEncodingException 编码异常
         */
        private void handle(byte[] bytes) throws UnsupportedEncodingException {
            String line;
            if (charSet == null){
                line = new String(bytes);
            }else {
                line = new String(bytes, charSet);
            }
            if (StringUtils.isNotBlank(line)){
                ihandle.handle(line);
                // 记录行数 */
                countLine.incrementAndGet();
            }
        }
    }

    /**
     * 该类用于存储每个分区的开始位置与结束位置
     * @author zwx
     */
    @Setter
    @Getter
    @ToString
    @NoArgsConstructor
    @AllArgsConstructor
    private static class StartEndPair{
        private Long start;
        private Long end;

        @Override
        public int hashCode() {
            final int prime = 11;
            int result = 30;
            result = prime * result + ((start == null) ? 0 : start.hashCode());
            result = prime * result + ((end == null) ? 0 : end.hashCode());
            return result;
        }

        @Override
        public boolean equals(Object obj) {
            if (this == obj) {
                return true;
            }
            if (obj == null) {
                return false;
            }
            if (obj instanceof StartEndPair){
                StartEndPair objSe = (StartEndPair) obj;
                return objSe.start.equals(this.start) && objSe.end.equals(this.end);
            }
            return false;
        }
    }

    /**
     * 构建一个bigFileReader
     * @author zwx
     */
    @Data
    @NoArgsConstructor
    @Accessors(chain = true)
    public static class Builder{
        private Integer bufferSize = 1024*1024;
        private String charSet;
        private Integer threadSize;
        private IHandle handle;
        private File file;

        public Builder(String filePath, IHandle handle){
            this.file = new File(filePath);
            if (!this.file.exists()){
                throw new IllegalArgumentException("文件不存在");
            }
            this.handle = handle;
        }

        public BigFileReader build(){
            return new BigFileReader(this.file, this.bufferSize, this.charSet, this.threadSize, this.handle);
        }
    }
}
